package main

import (
	"encoding/json"
	"strings"
	"time"

	"../logger"
	"../util"
	"github.com/Shopify/sarama"
)

type Producer struct {
	producer sarama.AsyncProducer
}

func NewProducer() (Producer, error) {
	// log
	// sarama.Logger = logger

	// config
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 60 * time.Second
	// config.Producer.RequiredAcks = sarama.WaitForAll
	// config.Producer.Partitioner = sarama.NewRandomPartitioner

	p, err := sarama.NewAsyncProducer(strings.Split("192.168.134.132:9092", ","), config)
	// producer, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)
	return Producer{producer: p}, err
}

func parseData(data string) ([]byte, error) {
	str := strings.Replace(data, "'", "\"", -1)
	dt := make([]map[string]string, 0)
	err := json.Unmarshal([]byte(str), &dt)
	if err != nil {
		logger.Errorf("Fail to parse data as json: %s\n", str)
		return nil, util.RequestArgsErr
	}
	return []byte(str), nil
}

func (p Producer) ProduceMsg(args Args) error {
	var data []byte
	var err error
	if data, err = parseData(args.data); err != nil {
		return err
	}

	// produce msg
	msg := &sarama.ProducerMessage{
		Timestamp: time.Now(),
		Topic:     "test",
		Key:       sarama.StringEncoder(args.table),
		Value:     sarama.ByteEncoder(data),
	}

	p.producer.Input() <- msg

	select {
	case err := <-p.producer.Errors():
		logger.Errorf("Send message error: %v\n", err)
		return err
	case msg := <-p.producer.Successes():
		key, err := msg.Key.Encode()
		if err != nil {
			logger.Errorf("Fail to encode msg.Key: %v\n", msg.Key)
		}
		val, err := msg.Value.Encode()
		if err != nil {
			logger.Errorf("Fail to encode msg.Value: %v\n", msg.Value)
		}
		if err == nil {
			timestamp := msg.Timestamp //.Format("2006-01-02 15:04:05")
			logmsg := "Send message success. Topic:%v, Key:%v, Value:%v, Timestamp:%v\n"
			logger.Printf(logmsg, msg.Topic, string(key), string(val), timestamp)
		}
	}
	return nil
}

func (p Producer) Close() error {
	return p.producer.Close()
}
